package com.one.flink;

import com.one.config.SyncConfig;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class FlinkMySQLToGaussSync {

    private static final SyncConfig config = SyncConfig.getInstance();
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) throws Exception {


        ParameterTool params = ParameterTool.fromArgs(args);


        log.info("开始初始化 MySQL 到 GaussDB 的同步任务...");

        // 1. 创建Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 配置checkpoint
        configureCheckpoint(env);

        // 3. 配置MySQL CDC Source
        MySqlSource<String> mySqlSource = configureMySqlSource(params);

        // 4. 创建数据流并处理
        createAndProcessDataStream(env, mySqlSource, params);

        // 5. 启动执行
        env.execute("MySQL to GaussDB Sync Job");
    }

    private static void configureCheckpoint(StreamExecutionEnvironment env) {
        // 配置checkpoint
        env.enableCheckpointing(config.getCheckpointInterval());
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // 配置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));

        log.info("Checkpoint配置完成: 间隔={}ms, 模式=EXACTLY_ONCE", config.getCheckpointInterval());
    }

    private static MySqlSource<String> configureMySqlSource(ParameterTool params) {


        Integer mysqlPort = Integer.valueOf(params.get("MYSQL_PORT"));
        String mysqlHost = params.get("MYSQL_HOST");
        String mysqlDatabase = params.get("MYSQL_DATABASE");
        String mysqlTable = params.get("MYSQL_TABLE");
        String mysqlPassword = params.get("MYSQL_PASSWORD");
        String mysqlUsername = params.get("MYSQL_USERNAME");
        String sourceTable = mysqlDatabase + "." + mysqlTable;
        log.info("配置 MySQL Source: {}", sourceTable);

        return MySqlSource.<String>builder()
                .hostname(mysqlHost)
                .port(mysqlPort)
                .databaseList(mysqlDatabase)
                .tableList(sourceTable)
                .username(mysqlUsername)
                .password(mysqlPassword)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial()) // 全量+增量
                .serverTimeZone("Asia/Shanghai")
                .build();
    }

    private static void createAndProcessDataStream(
            StreamExecutionEnvironment env,
            MySqlSource<String> mySqlSource,
            ParameterTool params) {

        log.info("创建数据流处理管道...");

        // 创建数据源
        DataStreamSource<String> stream = env.fromSource(
                mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "MySQL CDC Source"
        );

        // 转换并写入GaussDB
        stream.map(new RecordTransformer())
                .addSink(createGaussSink(params))
                .name("GaussDB Sink");

        log.info("数据流处理管道创建完成");
    }

    private static class RecordTransformer implements MapFunction<String, String> {
        @Override
        public String map(String value) throws Exception {
            try {
                JsonNode jsonNode = objectMapper.readTree(value);
                JsonNode after = jsonNode.get("after");

                if (after != null && after.has("name")) {
                    String name = after.get("name").asText();
                    String upperName = name.toUpperCase();
                    ((ObjectNode) after).put("name", upperName);
                    log.info("转换name: {} -> {}", name, upperName);
                }

                return objectMapper.writeValueAsString(jsonNode);
            } catch (Exception e) {
                log.error("数据转换失败: {}", value, e);
                throw e;
            }
        }
    }

    private static SinkFunction<String> createGaussSink(ParameterTool params) {


        String gaussUrl = params.get("GAUSS_URL");
        String gaussTable = params.get("GAUSS_TABLE");
        String gaussUsername = params.get("GAUSS_USERNAME");
        String gaussPassword = params.get("GAUSS_PASSWORD");

        log.info("配置 GaussDB Sink: {}", gaussUrl);

        String insertSql = String.format(
                "INSERT INTO %s (id, name, gender) VALUES (?, ?, ?) ",
                gaussTable
        );

        return JdbcSink.sink(
                insertSql,
                (statement, record) -> {
                    try {
                        JsonNode jsonNode = objectMapper.readTree(record);
                        JsonNode after = jsonNode.get("after");

                        if (after == null) {
                            log.warn("跳过无after数据的记录: {}", record);
                            return;
                        }

                        statement.setInt(1, after.get("id").asInt());
                        statement.setString(2, after.get("name").asText());
                        statement.setString(3, after.get("gender").asText());

                        log.debug("准备写入数据: id={}, name={}, gender={}",
                                after.get("id").asInt(),
                                after.get("name").asText(),
                                after.get("gender").asText());
                    } catch (Exception e) {
                        log.error("处理记录失败: {}", record, e);
                        throw new RuntimeException("处理记录失败", e);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(config.getBatchSize())
                        .withBatchIntervalMs(config.getBatchInterval())
                        .withMaxRetries(config.getMaxRetries())
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(gaussUrl)
                        .withDriverName("org.postgresql.Driver")
                        .withUsername(gaussUsername)
                        .withPassword(gaussPassword)
                        .build()
        );
    }
} 